package com.gin.cep;

import com.gin.cep.vo.LoginEvent;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * 10秒内连续登录失败3次
 *
 * @author gin
 * @date 2021/4/27
 */
public class LoginCepTest {

    public static void main(String[] args) {
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置为事件时间时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //数据流
        SingleOutputStreamOperator<LoginEvent> stream = env.fromCollection(new LinkedList<LoginEvent>() {
            {
                add(new LoginEvent(1, "张三", "fail", 1619008072000L));
                add(new LoginEvent(2, "张三", "fail", 1619008073000L));
                add(new LoginEvent(3, "张三", "fail", 1619008074000L));
                add(new LoginEvent(4, "李四", "fail", 1619008075000L));
                add(new LoginEvent(5, "李四", "success", 1619008076000L));
                add(new LoginEvent(6, "张三", "fail", 1619008077000L));
            }
        }).assignTimestampsAndWatermarks(//指定水印延迟3秒: Time.seconds(3)
                //使用flink实现好的: BoundedOutOfOrdernessTimestampExtractor
                new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                    //将数据中的时间字段提取出来，转成Long类型,不改变输入的数据样式
                    @Override
                    public long extractTimestamp(LoginEvent loginEvent) {
                        return loginEvent.getLoginTime();
                    }
                });

        KeyedStream<LoginEvent, String> keyedStream = stream.keyBy(new KeySelector<LoginEvent, String>() {
            @Override
            public String getKey(LoginEvent loginEvent) throws Exception {
                return loginEvent.getUserName();
            }
        });

        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("myStart")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return "fail".equals(loginEvent.loginType);
                    }
                })
                //连续失败3次
                .timesOrMore(3)
                .greedy()
                //10秒内
                .within(Time.seconds(10));

        PatternStream<LoginEvent> ps = CEP.pattern(keyedStream, pattern);

        SingleOutputStreamOperator<String> result = ps.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                StringBuilder sb = new StringBuilder();
                //定义了几个 begin , map中就有几条数据
                List<LoginEvent> myStart = map.get("myStart");
                sb.append("用户名:").append(myStart.get(0).userName);
                for (int i = 0; i < myStart.size(); i++) {
                    sb.append("; 第${").append(i+1).append("}次登录失败, id=")
                            .append(myStart.get(i).getId()).append(", time=")
                            .append(myStart.get(i).getLoginTime());
                }
                return sb.toString();
            }
        });

        result.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
